package ca.cbc.msb.mq;

import javax.jms.JMSException;
import javax.naming.NamingException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ca.cbc.msb.entity.MSBMessage;
import ca.cbc.msb.enums.QueueEnum;

import com.google.inject.Inject;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 *  runnable, JMS implementation of <code>MSBMessageConsumer</code>.
 *  
 * @author William Wei
 *
 */
public class MessageConsumerImpl implements MSBMessageConsumer{

//	private  ConsumeMessageUtil connector;
    @Inject
	private  MessageUtil connector;
	private MessageHandler handler;
	private QueueEnum queueName;
	private MSBMessage msgIn;
	private AtomicBoolean running=new AtomicBoolean(false);


    private Thread myThread;
	private Logger log = LoggerFactory.getLogger(this.getClass());
	
	public MessageConsumerImpl(QueueEnum queueName) throws NamingException, JMSException{
		this.queueName =queueName;
		connector = new MessageUtilImpl();
//		createHandler();
	}

    public MessageConsumerImpl(){
    }


	@Override
	public void run() {
        running.set(true);
		while(running.get()){
			try {
				msgIn = connector.getMessage(queueName);
			} catch (NamingException e) {
				log.error("JNDI lookup error for "+queueName,e);
			}  catch (Exception e) {
				log.error("Unknown Error when getting the message", e);
			}
			
			if(msgIn == null){
				continue;
			}
			
			try {
				handler.handle(msgIn, connector);
			} catch (NamingException e) {
				log.error("Error getting queue from JNDI: ",queueName,e);
			} catch (JMSException e) {
				log.error("Error when getting/sending  message to queue: "+queueName, e);
			} catch (Exception e) {
				log.error("Unknown Error when handling the message", e);
			}
		}		
		
		try {
            connector.disconnect();
            log.info("Message Consumer {} shutdown successfully", queueName);
		} catch (Exception e) {
			log.error("!!!Unknow Error when joining the message", e);
		} finally {
            log.info("closing the jms session for {} , ", queueName);
        }
	}

	public void setHandler(MessageHandler handler){
		this.handler = handler;
	}
	public MSBMessage getMsg() {
		return msgIn;
	}

	public void setMsg(MSBMessage msg) {
		this.msgIn = msg;
	}

	public void stop(){
         log.debug("Trying to stop {}", queueName);
		 running.set(false);
        try {
            myThread.join(30000);
        } catch (InterruptedException e) {
            log.error("Queue {} consumer refused to shutdown in 30s and got shot", queueName);
        }
    }

	@Override
	public void start() {
		myThread = new Thread(this);
        myThread.start();
		
	}

	
}

